DataformのREST APIを用いてDataformのGitリポジトリのファイルを複製するスクリプト

import os
import requests
import json
import base64

# 定数の定義
ENDPOINT = "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token"
PROJECT_ID = "PROJECT ID" #プロジェクトID
LOCATION = "asia-northeast1" #Dataformのリポジトリがあるリージョン
REPOSITORY = "リポジトリ名" #Dataformのリポジトリ名
BASE_PATH = "./cloned_repository"  # ローカルに作成するディレクトリのルート

def fetch_token():
    '''
    Token取得用関数
    指定されたエンドポイントからアクセストークンを取得します。
    '''
    token_header = {"Metadata-Flavor": "Google"}
    token_response = requests.get(ENDPOINT, headers=token_header)
    token_response.raise_for_status()  
    token_json_data = token_response.json()
    token = token_json_data["access_token"]
    return token

def query_directory_contents(project_id, location, repository, directory=""):
    '''
    ディレクトリ内容クエリ関数
    指定されたリポジトリ内のディレクトリ内容をクエリして一覧を取得します
    '''
    url = f"https://dataform.googleapis.com/v1beta1/projects/{project_id}/locations/{location}/repositories/{repository}:queryDirectoryContents"
    headers = {
        "Authorization": f"Bearer {fetch_token()}"
    }
    params = {
        "path": directory
    }
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    return response.json()

def read_file(project_id, location, repository, file_path):
    '''
    ファイル読み取り関数
    指定されたファイルの内容を取得します
    '''
    url = f"https://dataform.googleapis.com/v1beta1/projects/{project_id}/locations/{location}/repositories/{repository}:readFile"
    headers = {
        "Authorization": f"Bearer {fetch_token()}"
    }
    params = {
        "path": file_path
    }
    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()
    return response.json().get("contents", "")

def recursive_search(project_id, location, repository, directory=""):
    '''
    再帰的検索関数
    指定されたディレクトリ内のすべてのファイルとディレクトリを再帰的に検索します
    '''
    contents = query_directory_contents(project_id, location, repository, directory)
    all_entries = []

    for entry in contents.get("directoryEntries", []):
        if "file" in entry:
            all_entries.append(f"{directory}/{entry['file']}" if directory else entry['file'])
        elif "directory" in entry:
            sub_directory = f"{directory}/{entry['directory']}" if directory else entry['directory']
            all_entries.extend(recursive_search(project_id, location, repository, sub_directory))

    return all_entries

def create_local_file_structure(base_path, file_path, file_contents):
    '''
    ローカルファイル構造作成関数
    指定されたパスにファイルを作成し、その内容を書き込みます
    '''
    full_path = os.path.join(base_path, file_path)
    os.makedirs(os.path.dirname(full_path), exist_ok=True)
    with open(full_path, 'wb') as file:
        file.write(base64.b64decode(file_contents))

def main():
    '''
    メイン関数
    プロジェクトID、ロケーション、およびリポジトリ名を指定して、リポジトリ内のすべてのファイルをローカルに作成します
    '''
    all_entries = recursive_search(PROJECT_ID, LOCATION, REPOSITORY)
    for entry in all_entries:
        print(f"Processing file: {entry}")
        file_contents = read_file(PROJECT_ID, LOCATION, REPOSITORY, entry)
        if not file_contents:
            print(f"Warning: File contents are empty for {entry}")
        create_local_file_structure(BASE_PATH, entry, file_contents)
        print(f"Created file: {os.path.join(BASE_PATH, entry)}")

if __name__ == "__main__":
    main()

~$ python get_dataform_code.py 
Processing file: .gitignore
Created file: ./cloned_repository/.gitignore
Processing file: dataform.json
Created file: ./cloned_repository/dataform.json
・・・以下省略